package main

import (
	"context"
	"fmt"
	"github.com/MingOf/logDemo/conf"
	"github.com/MingOf/logDemo/kafka"
	"github.com/Shopify/sarama"
	"go.etcd.io/etcd/clientv3"
	"logagent/etcd"
	"logagent/taillog"
	"sync"
	"time"
)

var wg sync.WaitGroup

// logAgent 入口程序
func main() {
	var (
		producer   sarama.SyncProducer
		err        error
		newTail    *taillog.Tail
		cli        *clientv3.Client
		logEntries []*etcd.LogEntry
	)

	//filePath := cfg.Section("taillog").Key("path").String()
	//topic := cfg.Section("kafka").Key("topic").String()
	//addr := cfg.Section("kafka").Key("address").String()
	//cfg, err = ini.Load("./conf/config.ini")

	//0.从文件中获取ini配置信息
	cfg := conf.NewConfig("./conf/config.ini")
	//1.新建kafka producer
	if producer, err = kafka.NewProducer([]string{cfg.Kafka.Address}); err != nil {
		fmt.Println("[main] create Producer failed,err:", err)
		return
	}
	defer producer.Close()

	//2.新建etcd连接
	if cli, err = etcd.NewClient([]string{cfg.Etcd.Address}, time.Duration(cfg.Etcd.Timeout)*time.Second); err != nil {
		fmt.Println("[main] init etcd err:", err)
		return
	}
	defer cli.Close()

	//3.从etcd中获取日志收集项的配置信息
	if logEntries, err = etcd.GetConf(cli, cfg.Etcd.Key); err != nil {
		fmt.Println("[main] get logEntry err:", err)
		return
	}

	//4.遍历配置信息，将配置信息中指示的log文件的内容写入kafka
	for _, entry := range logEntries {
		fmt.Printf("[main] found entry: %s, found topic: %s\n", entry.Path, entry.Topic)
		ctx, cancel := context.WithCancel(context.Background())
		if newTail, err = taillog.NewTail(&taillog.TailContext{Ctx: ctx, Cancel: cancel}, entry.Path, entry.Topic); err != nil {
			fmt.Println("[main] NewTail err:", err)
		}
		go newTail.SendToKafka(producer)
	}
	wg.Add(1)
	//5.监视日志配置信息的变化
	go etcd.WatchAndServe(cli, cfg.Etcd.Key, producer, taillog.Diff)
	wg.Wait()

}
